const Redis = require('ioredis');
const { MessageBroker } = require('./index.js');

class RedisBroker extends MessageBroker {
    constructor() {
        super();
        this.pub = null;
        this.sub = null;
        this.handlers = new Map();
        this.workers = new Map(); // 使用 Map 存储消费者状态
        this.heartbeatInterval = 5000; // 心跳间隔时间
    }

    async start(redisUrl = 'redis://127.0.0.1:6379/0') {
        this.pub = new Redis(redisUrl);
        this.sub = new Redis(redisUrl);
        console.log(`## Redis Broker 已启动: ${redisUrl}...`);

        // 监听消息
        this.sub.on('message', (channel, message) => {
            // console.log(`sub channel = ${channel}, message = ${message}`);
            if (channel === 'heartbeat') {
                // 处理心跳消息
                const { workerId, content } = JSON.parse(message);
                this.handleHeartbeat(workerId, content);
            } else {
                // 处理其他消息
                const handlers = this.handlers.get(channel) || [];
                handlers.forEach((handler) => handler(JSON.parse(message)));
            }
        });

        // 订阅心跳频道
        await this.sub.subscribe('heartbeat');

        // 启动心跳检查
        this.startHeartbeatCheck();
    }

    // 启动心跳检查
    startHeartbeatCheck() {
        setInterval(() => {
            const now = Date.now();
            for (const [workerId, workerInfo] of this.workers.entries()) {
                if (now - workerInfo.lastHeartbeat > this.heartbeatInterval * 2) {
                    this.workers.delete(workerId);
                    console.log(`消费者 ${workerId} 已断开连接`);
                }
            }
        }, this.heartbeatInterval);
    }

    // 处理心跳消息
    handleHeartbeat(workerId, content) {
        this.workers.set(workerId, {
            address: content.ip,
            lastHeartbeat: Date.now(),
            status: 'connected',
            customInfo: content.customInfo
        });
    }

    async publish(taskType, worker, message) {
        if (!this.pub) {
            throw new Error('Redis Broker 未启动，请先调用 start 方法');
        }

        const channelName = `${taskType}:${worker}`
        try {
            await this.pub.xadd(channelName, '*', 'message', JSON.stringify(message));
            console.log(`消息已发布到 Stream, message = \n`, this.truncateMessage(message));
        } catch (error) {
            console.error('发布消息时出错:', error);
        }
    }

    async subscribe(taskType, callback) {
        if (!this.sub) {
            throw new Error('Redis Broker 未启动，请先调用 start 方法');
        }

        try {
            const handlers = this.handlers.get(taskType) || [];
            if (handlers.length === 0) {
                await this.sub.subscribe(taskType);
            }

            if (!handlers.includes(callback)) {
                handlers.push(callback);
                this.handlers.set(taskType, handlers);
            } else {
                console.warn(`回调函数已存在，无需重复订阅: ${taskType}`);
            }
        } catch (error) {
            console.error('订阅频道时出错:', error);
        }
    }

    async unsubscribe(taskType, callback) {
        if (!this.sub) {
            throw new Error('Redis Broker 未启动，请先调用 start 方法');
        }

        try {
            if (this.handlers.has(taskType)) {
                const handlers = this.handlers.get(taskType);
                const index = handlers.indexOf(callback);
                if (index !== -1) {
                    handlers.splice(index, 1);
                    if (handlers.length === 0) {
                        await this.sub.unsubscribe(taskType);
                        this.handlers.delete(taskType);
                        console.log(`已取消订阅任务类型: ${taskType}`);
                    } else {
                        this.handlers.set(taskType, handlers);
                    }
                } else {
                    console.warn(`未找到与任务类型 ${taskType} 关联的回调函数`);
                }
            } else {
                console.warn(`任务类型 ${taskType} 未被订阅`);
            }
        } catch (error) {
            console.error('取消订阅时出错:', error);
        }
    }

    async close() {
        if (!this.sub || !this.pub) {
            throw new Error('Redis Broker 未启动，请先调用 start 方法');
        }

        try {
            await this.sub.unsubscribe();
            await this.sub.quit();
            await this.pub.quit();
            this.handlers.clear();
            this.workers.clear(); // 清空消费者状态
            console.log('## Close Redis Broker OK!');
        } catch (error) {
            console.error('关闭 RedisBroker 时出错:', error);
        }
    }
}

module.exports = { RedisBroker };
